Pekerjaan Latar (Background Jobs) & Antrean Tugas

Background task (juga disebut background job) adalah kode, logika, atau alur kerja apa pun yang berjalan di luar siklus hidup request-response. Bayangkan interaksi HTTP khas: klien mengirim request, server Anda memprosesnya dan mengirim respons. Apa pun yang tidak perlu terjadi di dalam jendela sempit itu adalah kandidat background task.

Karakteristik kuncinya: ia tidak perlu sinkron. Ia tidak misi-kritis dalam arti klien sedang menunggunya sekarang. Karena itu, kita bisa dengan aman menggesernya ke proses terpisah yang menyelesaikannya dalam waktunya sendiri, sesuai cara ia diprogram.

Contoh motivasi klasiknya pendaftaran pengguna + verifikasi email. Ini gambaran utuh apa yang terjadi:

• 1. Pengguna mengisi form pendaftaran. Mereka memberi email, username, kata sandi. Frontend memvalidasi batasan dasar dan membuat panggilan API ke backend Anda.
• 2. Backend memvalidasi & menyimpan data. Kekuatan kata sandi, keunikan email, hashing, menulis record pengguna ke database, membuat token verifikasi sekali pakai (OTP 6 digit atau URL bertanda tangan).
• 3. Mengirim email verifikasi. Ini menuntut memanggil penyedia email pihak ketiga (Resend, Mailgun, Brevo). Backend Anda kini membuat request HTTP keluar ke layanan eksternal.
• 4. Penyedia email memproses dan mengirimkan. Penyedia memeriksa API key Anda, memvalidasi email, merutekan via SMTP, dan mengirimkan. Sukses/gagal dikembalikan.

Masalahnya: Pengiriman Email Sinkron

Jika langkah 3-4 terjadi sinkron di handler request Anda, Anda memperkenalkan ketergantungan keras pada layanan eksternal di dalam jalur kritis Anda. Dua skenario kegagalan:

Di kedua kasus, masalah fundamentalnya sama: Anda menggandeng keandalan API Anda pada uptime layanan pihak ketiga. Menggeser email ke background task memisahkan keduanya sepenuhnya.

Alur Sinkron (Blocking)

Alur Asinkron (Background)

Task queue adalah sistem untuk mengelola dan mendistribusikan background job. Ia mesin di balik layar yang memungkinkan pemrosesan latar yang andal dan terlepas. Anggap ia daftar tugas untuk backend Anda, aplikasi Anda menambahkan item ke daftar, dan worker mengambilnya satu per satu.

Serialisasi & Deserialisasi

Karena task menyeberangi batas proses, semua data harus diserialkan. JSON format paling umum. Worker men-deserialize kembali ke tipe natif:

// Task payload (JSON in the queue)
{
  "type": "send_verification_email",
  "user_id": "usr_01J2K...",
  "email": "alice@example.com",
  "token": "eyJhbGci...",
  "created_at": "2024-01-15T10:30:00Z"
}
JSON

Sisi Producer

Tanggung jawab producer minimal dan cepat, ia tidak boleh melakukan kerja berat. Tugasnya saja:

1. Kumpulkan semua data yang dibutuhkan consumer untuk melakukan task.
2. Serialkan data itu ke format portabel (JSON).
3. Dorong payload terserialisasi ke antrean.
4. Kembali, seketika. Jangan menunggu.

Sisi Consumer / Worker

Consumer berjalan sebagai proses terpisah: entah di codebase yang sama (entry-point/biner berbeda) atau layanan yang sepenuhnya terpisah. Ia:

1. Mengonfigurasi antrean mana yang didengarkan: antrean berbeda untuk jenis tugas berbeda (antrean email, antrean notifikasi, antrean pemrosesan gambar, dll.).
2. Mendaftarkan handler: untuk tiap jenis tugas, sebuah fungsi didaftarkan. Saat tugas itu di-dequeue, fungsi itu dipanggil.
3. Polling atau didorong tugas: tergantung broker (Redis/RabbitMQ memakai long-polling; SQS memakai interval polling).
4. Men-deserialize payload: JSON -> struct/dict/objek natif.
5. Menjalankan handler: kerja sebenarnya: memanggil API email, mengubah ukuran gambar, dll.
6. Mengakui (ACK) sukses: memberi tahu antrean tugas selesai dan bisa dihapus.
7. Melaporkan kegagalan (NACK): saat error, antrean tahu untuk mencoba ulang.

"Antrean"-nya didukung infrastruktur sungguhan, sebuah message broker. Ini pilihan utamanya:

Internal Redis: Cara Ia Menyimpan Tugas

Redis menawarkan dua struktur data berbeda yang dipakai library task queue di balik layar. Memahami mana yang dipakai framework Anda, dan kenapa, membantu Anda mengoperasikannya dengan benar di produksi.

Redis Lists (dipakai: Celery, BullMQ, Sidekiq)

Redis List adalah doubly-linked list berisi string. Task queue memakai dua perintah: LPUSH (producer mendorong ke kiri/kepala) dan BRPOP (worker memblokir menunggu pop dari kanan/ekor). Ini memberi perilaku FIFO klasik.

# What Celery does under the hood in Redis:

# Producer (your API), pushes serialized task to left of list
LPUSH celery '{"task":"send_email","args":["alice@example.com","token123"]}'

# Worker, blocks waiting for item, pops from right (FIFO)
# BRPOP blocks for up to 5 seconds, then re-polls
BRPOP celery 5

# To see queue depth at any time:
LLEN celery   # -> 42 (42 pending tasks)
Redis CLI

Redis Streams (dipakai: Asynq, arsitektur baru)

Redis Streams (ditambahkan di Redis 5.0) adalah log append-only, secara konsep mirip Apache Kafka tapi di Redis. Ia dirancang khusus untuk memecahkan masalah Redis Lists untuk antrean pesan.

# Producer adds entry to stream
# * means auto-generate ID (timestamp-sequence: 1705312200000-0)
XADD myapp:email_queue * user_id usr_01J2K email alice@example.com token eyJ...

# Create a consumer group (workers belong to a group)
XGROUP CREATE myapp:email_queue email_workers $ MKSTREAM

# Worker reads next undelivered message (> means "new")
XREADGROUP GROUP email_workers worker-1 COUNT 1 BLOCK 5000 STREAMS myapp:email_queue >

# After successful processing, ACK the message by ID
XACK myapp:email_queue email_workers 1705312200000-0

# Check pending (delivered but not ACKed), these are in-flight tasks
XPENDING myapp:email_queue email_workers - + 10
Redis CLI

Amazon SQS adalah layanan antrean pesan terkelola penuh milik AWS. Tanpa server yang dikelola, skala otomatis, dan mereplikasi lintas beberapa availability zone. Ia punya dua mode dengan jaminan yang berbeda fundamental.

Cara Kerja Visibility Timeout SQS (Khusus AWS)

SQS mengimplementasikan visibility timeout secara natif. Saat worker memanggil ReceiveMessage, pesan menjadi tak terlihat bagi semua consumer lain selama timeout terkonfigurasi (default 30 detik, maksimal 12 jam). Worker harus memanggil DeleteMessage saat sukses, atau tidak melakukan apa pun untuk membiarkannya terlihat lagi untuk retry.

// Go, SQS producer + consumer using AWS SDK v2
package main

import (
    "context"
    "encoding/json"
    "github.com/aws/aws-sdk-go-v2/aws"
    "github.com/aws/aws-sdk-go-v2/service/sqs"
)

// -- PRODUCER ------------------------------------------
func EnqueueEmailTask(ctx context.Context, client *sqs.Client, queueURL string, payload EmailPayload) error {
    body, _ := json.Marshal(payload)
    _, err := client.SendMessage(ctx, &sqs.SendMessageInput{
        QueueUrl:    aws.String(queueURL),
        MessageBody: aws.String(string(body)),
        // For FIFO queue: add MessageGroupId + MessageDeduplicationId
        // MessageGroupId:         aws.String("email-group"),
        // MessageDeduplicationId: aws.String(payload.UserID),
    })
    return err
}

// -- CONSUMER ------------------------------------------
func PollQueue(ctx context.Context, client *sqs.Client, queueURL string) {
    for {
        result, _ := client.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
            QueueUrl:            aws.String(queueURL),
            MaxNumberOfMessages: 10,          // batch up to 10
            WaitTimeSeconds:     20,          // long-polling reduces empty receives
            VisibilityTimeout:   60,          // 60s to process before redelivery
        })

        for _, msg := range result.Messages {
            if err := processMessage(ctx, msg); err == nil {
                // ACK: delete from queue on success
                client.DeleteMessage(ctx, &sqs.DeleteMessageInput{
                    QueueUrl:      aws.String(queueURL),
                    ReceiptHandle: msg.ReceiptHandle, // unique handle per receive
                })
            }
            // On error: do nothing, visibility timeout expires, SQS redelivers
        }
    }
}
Go / AWS SDK v2

Saat task gagal (handler melempar exception, API eksternal mengembalikan 5xx, timeout jaringan), framework meng-antrekan ulang task untuk retry. Mengulang seketika secara naif akan menghantam layanan yang sudah kewalahan. Solusinya: exponential backoff.

Algoritma Exponential Backoff

Setelah tiap kegagalan, waktu tunggu sebelum retry berikutnya berlipat dua. Formula umum:

// Retry delay formula
delay = base_delay * (2 ^ attempt_number) + jitter

// Example with base_delay=1min, max_retries=5:
Attempt 1 -> wait 1 min   -> retry
Attempt 2 -> wait 2 min   -> retry
Attempt 3 -> wait 4 min   -> retry
Attempt 4 -> wait 8 min   -> retry
Attempt 5 -> wait 16 min  -> DEAD LETTER QUEUE
PSEUDOCODE

Kenapa Jitter Penting

Tanpa jitter (kebisingan acak yang ditambahkan ke jeda), semua consumer mencoba ulang di waktu yang sama persis setelah kegagalan, menciptakan thundering herd yang memukul layanan eksternal dengan lonjakan. Menambah offset acak kecil menyebarkan beban.

Dead Letter Queue (DLQ)

Setelah semua retry habis, task dipindahkan ke Dead Letter Queue. Ini antrean terpisah untuk task yang gagal terus-menerus. Engineer bisa memeriksa pesan DLQ, mendiagnosis akar masalahnya, dan memutar ulang secara manual setelah masalah diperbaiki.

Saat consumer mengambil task (dequeue), task tidak langsung dihapus. Gantinya, ia menjadi tak terlihat bagi consumer lain selama durasi terkonfigurasi, visibility timeout. Ini periode tugas dianggap "sedang berjalan".

Kenapa Ini Ada?

Tanpa visibility timeout, jika consumer crash di tengah task, task akan hilang permanen, ia sudah dihapus dari antrean saat consumer mengambilnya. Visibility timeout memastikan jika worker tidak mengakui dalam jendela (ia crash, menggantung, atau layanan eksternal timeout), task jadi terlihat lagi dan worker lain bisa mengambilnya. Task tidak pernah hilang.

Chain Task: Contoh Upload Video

Notifikasi Push: Cara Kerja Sebenarnya

Backend Anda tidak bisa mengirim notifikasi push langsung ke ponsel. Ini alur lengkapnya:

1. Pengguna menginstal aplikasi -> OS membuat device token unik (token Apple APNs / token Google FCM).
2. Aplikasi mengirim token itu ke backend Anda saat peluncuran pertama, Anda menyimpannya per-pengguna di DB.
3. Saat ingin memberi tahu pengguna: antrekan task "kirim notifikasi push" dengan device token + payload pesan pengguna.
4. Worker dequeue -> memanggil API Apple APNs atau Google FCM dengan token dan payload.
5. Apple/Google mengirimkan notifikasi ke perangkat lewat kanal push tingkat OS.

Python: Celery + Redis

Celery adalah task queue Python paling populer. Memakai Redis (atau RabbitMQ) sebagai broker.

# tasks.py, Define the task (consumer side)
from celery import Celery
import resend

# Connect Celery to Redis broker
app = Celery('tasks', broker='redis://localhost:6379/0')

# Decorate the function as a Celery task
@app.task(bind=True, max_retries=5, default_retry_delay=60)
def send_verification_email(self, user_id: str, email: str, token: str):
    """
    Called by the worker process.
    self.retry() implements exponential backoff automatically.
    """
    try:
        params = {
            "from": "noreply@myapp.com",
            "to": [email],
            "subject": "Verify your email",
            "html": build_email_template(token),
        }
        resend.Emails.send(params)
    except Exception as exc:
        # Exponential backoff: 60s, 120s, 240s, 480s, 960s
        countdown = 60 * (2 ** self.request.retries)
        raise self.retry(exc=exc, countdown=countdown)
Python

# signup.py, Producer side (inside your API handler)
from fastapi import FastAPI
from tasks import send_verification_email

app = FastAPI()

@app.post("/signup")
async def signup(body: SignupRequest):
    # 1. Validate, hash password, save user to DB
    user = await create_user(body)
    token = generate_verification_token(user.id)

    # 2. Enqueue, returns instantly, does NOT wait for email
    send_verification_email.delay(
        user_id=user.id,
        email=user.email,
        token=token
    )

    # 3. Return 201 immediately
    return {"message": "Account created. Check your email."}, 201
Python / FastAPI

Go: Asynq + Redis

Asynq adalah library task queue terdistribusi yang sederhana dan andal untuk Go, didukung Redis Streams.

// tasks/email.go, Task definitions (producer + payload)
package tasks

import (
    "context"
    "encoding/json"
    "fmt"
    "github.com/hibiken/asynq"
)

const TypeSendVerificationEmail = "email:send_verification"

// Payload struct, serialized to JSON in queue
type EmailPayload struct {
    UserID string `json:"user_id"`
    Email  string `json:"email"`
    Token  string `json:"token"`
}

// NewSendVerificationEmailTask creates the Asynq task
func NewSendVerificationEmailTask(userID, email, token string) (*asynq.Task, error) {
    payload, err := json.Marshal(EmailPayload{UserID: userID, Email: email, Token: token})
    if err != nil {
        return nil, fmt.Errorf("json.Marshal: %w", err)
    }
    // asynq.MaxRetry, after 5 failures, moves to dead letter
    return asynq.NewTask(TypeSendVerificationEmail, payload, asynq.MaxRetry(5)), nil
}
Go

// handlers/email_handler.go, Consumer handler
package handlers

import (
    "context"
    "encoding/json"
    "fmt"
    "log/slog"
    tasks "myapp/tasks"
    "github.com/hibiken/asynq"
)

type EmailHandler struct {
    emailSvc EmailService // injected dependency
}

// HandleSendVerificationEmail, registered with the Asynq server
func (h *EmailHandler) HandleSendVerificationEmail(
    ctx context.Context,
    t *asynq.Task,
) error {
    // 1. Deserialize JSON -> Go struct
    var p tasks.EmailPayload
    if err := json.Unmarshal(t.Payload(), &p); err != nil {
        // Non-retryable error: mark as failed immediately
        return fmt.Errorf("json.Unmarshal: %w: %w", err, asynq.SkipRetry)
    }

    // 2. Execute: call email provider API
    slog.Info("sending verification email", "user_id", p.UserID, "email", p.Email)
    if err := h.emailSvc.SendVerification(ctx, p.Email, p.Token); err != nil {
        // Returning error triggers retry with exponential backoff
        return fmt.Errorf("emailSvc.SendVerification: %w", err)
    }

    // 3. Returning nil -> ACK to queue (task done)
    return nil
}
Go

// main.go, Wire producer and start worker server
package main

import (
    "github.com/hibiken/asynq"
    "myapp/handlers"
    "myapp/tasks"
)

func main() {
    redisOpt := asynq.RedisClientOpt{Addr: "localhost:6379"}

    // -- PRODUCER -------------------------------------
    client := asynq.NewClient(redisOpt)
    defer client.Close()

    task, _ := tasks.NewSendVerificationEmailTask(
        "usr_01J2K", "alice@example.com", "eyJhbGci...",
    )
    // Enqueue, returns immediately
    info, _ := client.Enqueue(task)
    // info.ID, info.Queue, info.State can be used for monitoring

    // -- CONSUMER (worker server) ---------------------
    srv := asynq.NewServer(redisOpt, asynq.Config{
        Concurrency: 10, // 10 concurrent workers
        Queues: map[string]int{
            "email":    6, // higher priority
            "default":  3,
            "low":      1,
        },
    })

    mux := asynq.NewServeMux()
    mux.HandleFunc(tasks.TypeSendVerificationEmail,
        handlers.EmailHandler{}.HandleSendVerificationEmail)

    srv.Run(mux)
}
Go

Go: Task Berulang dengan Asynq Scheduler

// scheduler/main.go, Cron-style recurring tasks
package main

import (
    "log"
    "github.com/hibiken/asynq"
)

func main() {
    scheduler := asynq.NewScheduler(
        asynq.RedisClientOpt{Addr: "localhost:6379"},
        nil,
    )

    // Send weekly report every Sunday at midnight
    scheduler.Register("0 0 * * 0",
        asynq.NewTask("report:weekly", nil))

    // Cleanup orphan sessions every 1st of the month
    scheduler.Register("0 3 1 * *",
        asynq.NewTask("sessions:cleanup", nil))

    if err := scheduler.Run(); err != nil {
        log.Fatal(err)
    }
}
Go

Python: Celery Beat (Task Berulang)

# celery_config.py, Celery Beat schedule
from celery.schedules import crontab

CELERYBEAT_SCHEDULE = {
    # Send weekly report every Sunday midnight
    'weekly-report': {
        'task': 'tasks.send_weekly_report',
        'schedule': crontab(hour=0, minute=0, day_of_week='sunday'),
    },
    # Cleanup orphan sessions on the 1st of each month
    'session-cleanup': {
        'task': 'tasks.cleanup_orphan_sessions',
        'schedule': crontab(hour=3, minute=0, day_of_month='1'),
    },
}
Python / Celery Beat

Saat platform Anda tumbuh ke ribuan pengguna serentak, setup task queue naif yang bekerja di pengembangan mulai retak. Inilah area yang harus Anda rekayasa hati-hati sebelum mencapai skala produksi.

Idempotensi berarti menjalankan task sekali atau 100 kali menghasilkan efek samping identik. Ini harga mati di sistem mana pun yang me-retry task, dan semua task queue produksi melakukannya. Ada beberapa pola konkret untuk mencapainya.

Pola 1: Idempotency Key

Berikan key unik ke tiap task saat dibuat (biasanya UUID atau hash dari input). Sebelum melakukan kerja apa pun, handler memeriksa apakah key itu sudah pernah diproses sukses. Jika ya, ia kembali lebih awal tanpa mengulang pekerjaan.

# Python, Idempotency key pattern with Redis
import redis
import hashlib

r = redis.Redis(host='localhost', port=6379)

@app.task(bind=True, max_retries=5)
def send_verification_email(self, user_id: str, email: str, token: str):
    # Build idempotency key from task inputs
    key = f"idem:verify_email:{user_id}:{token}"

    # SET NX = set only if Not eXists; EX = expire after 1 hour
    # Returns True if we are the FIRST execution, False if already done
    acquired = r.set(key, "done", nx=True, ex=3600)
    if not acquired:
        # Already processed, skip silently (idempotent return)
        return {"status": "already_sent"}

    # First time, actually send the email
    try:
        _send_email_via_provider(email, token)
    except Exception as exc:
        # Delete the key so next retry can attempt again
        r.delete(key)
        raise self.retry(exc=exc, countdown=60 * 2 ** self.request.retries)
Python

Pola 2: Transaksi Database + Upsert

Untuk task yang menulis ke database, pakai upsert (INSERT ... ON CONFLICT DO NOTHING / DO UPDATE) alih-alih INSERT polos. Dengan begitu, me-retry task cukup mencoba menyisipkan lagi dan diam-diam melewati duplikatnya.

-- PostgreSQL, Idempotent email log via upsert
-- If the verification_emails row already exists for this token, do nothing
INSERT INTO verification_emails (token, user_id, sent_at, provider_msg_id)
VALUES ($1, $2, NOW(), $3)
ON CONFLICT (token) DO NOTHING;

-- For updates: ON CONFLICT DO UPDATE
INSERT INTO user_status (user_id, email_verified, updated_at)
VALUES ($1, true, NOW())
ON CONFLICT (user_id) DO UPDATE
  SET email_verified = EXCLUDED.email_verified,
      updated_at = EXCLUDED.updated_at;
SQL

Pola 3: Rollback Transaksi Penuh (untuk task kompleks)

Untuk task bertahap banyak (seperti penghapusan akun) yang menyentuh banyak tabel, bungkus seluruh task dalam transaksi database. Jika ada langkah yang gagal, transaksi di-rollback sepenuhnya, meninggalkan database dalam state bersih untuk percobaan retry berikutnya.

// Go, Idempotent multi-step task with full transaction rollback
func (h *AccountHandler) HandleDeleteAccount(ctx context.Context, t *asynq.Task) error {
    var p DeleteAccountPayload
    json.Unmarshal(t.Payload(), &p)

    // Check if already deleted (idempotency guard)
    exists, _ := h.db.UserExists(ctx, p.UserID)
    if !exists {
        return nil // already deleted on a previous attempt, ACK cleanly
    }

    // Wrap all DB writes in a single transaction
    return h.db.WithTransaction(ctx, func(tx DB) error {
        steps := []func() error{
            func() error { return tx.DeleteUserProjects(ctx, p.UserID) },
            func() error { return tx.DeleteUserSessions(ctx, p.UserID) },
            func() error { return tx.DeleteUserAssets(ctx, p.UserID) },
            func() error { return tx.DeleteUserAccount(ctx, p.UserID) },
        }
        for _, step := range steps {
            if err := step(); err != nil {
                return err // triggers full rollback; task retried from scratch
            }
        }
        return nil // all steps succeeded -> commit -> ACK
    })
}
Go

Pola 4: Exactly-Once via Deduplikasi FIFO SQS

Antrean FIFO SQS menerima MessageDeduplicationId. Pesan apa pun dengan ID yang sama yang dikirim dalam jendela 5 menit dijatuhkan diam-diam. Ini idempotensi tingkat broker, handler bahkan tidak pernah melihat duplikatnya.

// Go, SQS FIFO with content-based deduplication
client.SendMessage(ctx, &sqs.SendMessageInput{
    QueueUrl:               aws.String("https://sqs.us-east-1.amazonaws.com/123/orders.fifo"),
    MessageBody:            aws.String(payload),
    MessageGroupId:         aws.String("order-processing"), // ordering group
    MessageDeduplicationId: aws.String(orderID),           // unique per business event
})
// Sending the same orderID twice within 5 min -> second is silently dropped by SQS
Go / AWS SQS FIFO

API eksternal memberlakukan rate limit. Jika worker Anda mengirim 500 email per detik dan limit Resend 100/detik, Anda akan kebanjiran error 429, retry terbuang, dan deliverability menurun. Anda perlu men-throttle worker agar tetap dalam batas.

Dua Algoritma Inti

Token Bucket di Python (Redis-backed)

# rate_limiter.py, Redis token bucket implementation
import time
import redis

class TokenBucketRateLimiter:
    def __init__(self, redis_client, key: str, capacity: int, refill_rate: float):
        self.r = redis_client
        self.key = key               # e.g. "ratelimit:resend_emails"
        self.capacity = capacity     # max burst (e.g. 100)
        self.refill_rate = refill_rate  # tokens per second (e.g. 10)

    def acquire(self) -> bool:
        """Returns True if a token was acquired, False if rate limited."""
        now = time.time()
        pipe = self.r.pipeline()

        # Lua script ensures atomic read-modify-write (no race conditions)
        lua_script = """
        local key = KEYS[1]
        local capacity = tonumber(ARGV[1])
        local refill_rate = tonumber(ARGV[2])
        local now = tonumber(ARGV[3])

        local bucket = redis.call('HMGET', key, 'tokens', 'last_refill')
        local tokens = tonumber(bucket[1]) or capacity
        local last_refill = tonumber(bucket[2]) or now

        -- Refill tokens based on time elapsed
        local elapsed = now - last_refill
        tokens = math.min(capacity, tokens + elapsed * refill_rate)

        if tokens >= 1 then
            tokens = tokens - 1
            redis.call('HMSET', key, 'tokens', tokens, 'last_refill', now)
            redis.call('EXPIRE', key, 3600)
            return 1
        end
        return 0
        """
        result = self.r.eval(lua_script, 1, self.key, self.capacity, self.refill_rate, now)
        return bool(result)

# Usage inside a Celery task
limiter = TokenBucketRateLimiter(r, "ratelimit:resend", capacity=100, refill_rate=10)

@app.task(bind=True, max_retries=10)
def send_email_task(self, email, token):
    if not limiter.acquire():
        # Rate limited, retry after a short delay (not counted as failure)
        raise self.retry(countdown=1, max_retries=60)  # retry every 1s up to 60 times
    _do_send_email(email, token)
Python

Sliding Window Rate Limiter di Go

// rate/sliding_window.go, Redis sorted-set sliding window
package rate

import (
    "context"
    "fmt"
    "time"
    "github.com/redis/go-redis/v9"
)

type SlidingWindowLimiter struct {
    client  *redis.Client
    key     string
    limit   int           // max requests
    window  time.Duration // time window
}

func (l *SlidingWindowLimiter) Allow(ctx context.Context) (bool, error) {
    now := time.Now()
    windowStart := now.Add(-l.window).UnixMicro()
    nowMicro := now.UnixMicro()

    pipe := l.client.Pipeline()
    // Remove entries older than the window
    pipe.ZRemRangeByScore(ctx, l.key, "0", fmt.Sprintf("%d", windowStart))
    // Count entries in current window
    countCmd := pipe.ZCard(ctx, l.key)
    // Add current request timestamp as a member
    pipe.ZAdd(ctx, l.key, redis.Z{Score: float64(nowMicro), Member: nowMicro})
    pipe.Expire(ctx, l.key, l.window*2)
    pipe.Exec(ctx)

    count := countCmd.Val()
    return count < int64(l.limit), nil
}

// Usage in Asynq handler
func (h *EmailHandler) HandleSendEmail(ctx context.Context, t *asynq.Task) error {
    allowed, _ := h.limiter.Allow(ctx)
    if !allowed {
        // Return a retryable error, asynq will retry with backoff
        return fmt.Errorf("rate limit exceeded: %w", asynq.ErrRetry)
    }
    return h.doSendEmail(ctx, t)
}
Go

Background task gagal diam-diam kecuali Anda membangun observabilitas ke dalam sistem. Stack produksi standarnya: Prometheus (koleksi & penyimpanan metrik) + Grafana (dashboard & alerting). Ini setup lengkapnya.

Ringkasan Arsitektur

Langkah 1: Instrumentasi Worker Anda

Setiap worker mengekspos endpoint HTTP /metrics dalam format teks Prometheus. Library prometheusclient (Python) atau prometheus/clientgolang (Go) menangani ini otomatis.

# Python, Full metrics instrumentation for a worker process
from prometheus_client import (
    Counter, Histogram, Gauge, start_http_server
)
import time

# -- Define metrics ---------------------------------------------
TASK_TOTAL = Counter(
    'worker_tasks_total',
    'Total tasks processed',
    ['task_name', 'status']  # labels: task_name, status=success|failure
)

TASK_DURATION = Histogram(
    'worker_task_duration_seconds',
    'Task processing duration',
    ['task_name'],
    buckets=[.01, .05, .1, .5, 1, 5, 10, 30]  # histogram bucket boundaries
)

QUEUE_DEPTH = Gauge(
    'worker_queue_depth',
    'Current tasks waiting in queue',
    ['queue_name']
)

RETRY_COUNT = Counter(
    'worker_task_retries_total',
    'Number of task retries',
    ['task_name']
)

# -- Decorator to wrap any task with metrics --------------------
def track_metrics(task_name: str):
    def decorator(func):
        def wrapper(*args, **kwargs):
            start = time.time()
            try:
                result = func(*args, **kwargs)
                TASK_TOTAL.labels(task_name=task_name, status="success").inc()
                return result
            except Exception as e:
                TASK_TOTAL.labels(task_name=task_name, status="failure").inc()
                raise
            finally:
                TASK_DURATION.labels(task_name=task_name).observe(time.time() - start)
        return wrapper
    return decorator

# Start /metrics server on port 9090 (Prometheus scrapes this)
start_http_server(9090)

# -- Apply to task ----------------------------------------------
@app.task(bind=True, max_retries=5)
@track_metrics("send_verification_email")
def send_verification_email(self, user_id, email, token):
    if self.request.retries > 0:
        RETRY_COUNT.labels(task_name="send_verification_email").inc()
    _do_send_email(email, token)
Python

Langkah 2: Konfigurasi Prometheus

# prometheus.yml, Scrape config
global:
  scrape_interval: 15s
  evaluation_interval: 15s

rule_files:
  - "alert_rules.yml"

alerting:
  alertmanagers:
    - static_configs:
        - targets: ["alertmanager:9093"]

scrape_configs:
  - job_name: "celery_workers"
    static_configs:
      - targets:
          - "worker-1:9090"
          - "worker-2:9090"
          - "worker-3:9090"

  - job_name: "redis"
    static_configs:
      - targets: ["redis-exporter:9121"]  # oliver006/redis_exporter

  - job_name: "api_server"
    static_configs:
      - targets: ["api:8080"]
YAML

Langkah 3: Aturan Alert

# alert_rules.yml, Fire alerts when things go wrong
groups:
  - name: task_queue_alerts
    rules:

      # Alert if queue depth stays high for 5 minutes
      - alert: HighQueueDepth
        expr: worker_queue_depth{queue_name="email"} > 1000
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Email queue backed up ({{ $value }} tasks)"
          description: "Queue depth has been above 1000 for 5+ minutes. Scale workers."

      # Alert if failure rate > 10% over last 5 minutes
      - alert: HighTaskFailureRate
        expr: |
          rate(worker_tasks_total{status="failure"}[5m])
          / rate(worker_tasks_total[5m]) > 0.1
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "Task failure rate above 10%"

      # Alert if a worker process goes down (no scrape for 1 min)
      - alert: WorkerDown
        expr: up{job="celery_workers"} == 0
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "Worker {{ $labels.instance }} is down"
YAML

Langkah 4: Panel Grafana Kunci yang Dibuat

1. Jaga Task Kecil dan Fokus

Satu task harus melakukan satu hal. Jangan menggabungkan pengiriman email + pembersihan DB + pelaporan analitik dalam satu task. Jika satu langkah gagal, Anda me-retry semuanya. Task kecil = retry presisi, debugging lebih mudah, skalabilitas lebih baik.

Jika banyak hal saling bergantung, pakai chain task (induk-anak). Jika independen, jalankan paralel.

2. Hindari Task Berumur Panjang

Jika sebuah task butuh menit untuk berjalan, pecah jadi potongan lebih kecil. Task panjang mengonsumsi thread worker, lebih sulit di-retry dengan benar, dan rawan kedaluwarsa visibility timeout. Task yang butuh 10 menit -> 10 task yang masing-masing 1 menit.

3. Penanganan Error & Logging yang Kokoh

Tidak seperti request web yang Anda bisa kembalikan 500 dan klien melihatnya, background task gagal diam-diam kecuali Anda membangun observabilitas. Setiap handler harus:

• Tangkap semua exception, jangan biarkan panic/exception tak tertangani mematikan worker diam-diam.
• Log task ID, user ID, jenis error, dan stack trace penuh saat kegagalan.
• Bedakan error sementara vs permanen, pakai SkipRetry (Asynq) atau raises=(SkipRetry,) (Celery) untuk error yang tak bisa dipulihkan.
• Kirim kegagalan kritis ke kanal alerting (Slack, PagerDuty).

4. Pantau Panjang Antrean & Kesehatan Worker

Set ambang alerting: jika kedalaman antrean > 10.000 -> picu auto-scaling atau alert engineer. Jika pod worker restart lebih dari 3 kali dalam 5 menit -> alert. Pakai dashboard Grafana yang didukung Prometheus (atau CloudWatch untuk SQS).

5. Pakai Dead Letter Queue

Jangan pernah biarkan task menghilang diam-diam. Konfigurasi DLQ supaya task yang menghabiskan semua retry mendarat di tempat yang bisa diperiksa. Audit DLQ secara teratur, kegagalan berulang di sana menandakan bug sistemik.

Task queue standar (Celery, BullMQ, Asynq) hebat untuk task independen, kirim-dan-lupakan, retry saat gagal. Tapi saat Anda punya workflow multi-langkah yang berumur panjang dengan ketergantungan kompleks, percabangan kondisional, langkah persetujuan manusia, atau logika kompensasi (pola saga), Anda butuh sesuatu yang lebih kuat: mesin orkestrasi workflow.

Task Queue vs Mesin Workflow

Konsep Inti Temporal

Contoh: Workflow Pemrosesan Video di Go

Chain task upload video yang sama dari Bagian 9, kini diimplementasikan dengan Temporal:

// workflows/video_processing.go
package workflows

import (
    "time"
    "go.temporal.io/sdk/workflow"
    "go.temporal.io/sdk/activity"
    "myapp/activities"
)

// VideoProcessingWorkflow, orchestrates the entire pipeline
// Temporal persists every step, crash at any point = resume here
func VideoProcessingWorkflow(ctx workflow.Context, videoID string) error {
    // Activity options, each step has its own retry policy
    actOpts := workflow.ActivityOptions{
        StartToCloseTimeout: 10 * time.Minute,
        RetryPolicy: &temporal.RetryPolicy{
            MaxAttempts:        3,
            InitialInterval:    time.Minute,
            BackoffCoefficient: 2.0,
        },
    }
    ctx = workflow.WithActivityOptions(ctx, actOpts)

    // Step 1: Encode video to multiple resolutions
    // If server crashes here, workflow resumes from step 1 on restart
    var encodedPath string
    if err := workflow.ExecuteActivity(ctx, activities.EncodeVideo, videoID).Get(ctx, &encodedPath); err != nil {
        return err // step 1 failed all retries, workflow fails
    }

    // Step 2 + 3: Thumbnail generation AND transcription in parallel
    thumbFuture := workflow.ExecuteActivity(ctx, activities.GenerateThumbnails, encodedPath)
    transcriptFuture := workflow.ExecuteActivity(ctx, activities.GenerateTranscription, encodedPath)

    // Wait for both, if either fails, the workflow fails (with its own retries first)
    if err := thumbFuture.Get(ctx, nil); err != nil { return err }
    if err := transcriptFuture.Get(ctx, nil); err != nil { return err }

    // Step 4: Process thumbnails (depends on step 2 completing)
    if err := workflow.ExecuteActivity(ctx, activities.ProcessThumbnailImages, videoID).Get(ctx, nil); err != nil {
        return err
    }

    // Step 5: Notify user their video is ready
    return workflow.ExecuteActivity(ctx, activities.NotifyUserVideoReady, videoID).Get(ctx, nil)
}
Go / Temporal

Contoh: Human-in-the-Loop dengan Signal (Go)

// workflows/payment_approval.go, Wait for human signal
func PaymentApprovalWorkflow(ctx workflow.Context, orderID string) error {
    // Register a signal channel, external events can wake this workflow
    approvalCh := workflow.GetSignalChannel(ctx, "approval-signal")

    // Step 1: Process payment details
    workflow.ExecuteActivity(ctx, activities.PreparePayment, orderID).Get(ctx, nil)

    // Step 2: Wait up to 24 hours for human approval
    // Temporal pauses here, no CPU/memory consumed while waiting
    var approved bool
    selector := workflow.NewSelector(ctx)
    selector.AddReceive(approvalCh, func(ch workflow.ReceiveChannel, more bool) {
        ch.Receive(ctx, &approved)
    })
    selector.AddFuture(workflow.NewTimer(ctx, 24*time.Hour), func(f workflow.Future) {
        approved = false // timeout, auto-reject
    })
    selector.Select(ctx)

    if !approved {
        // Saga: compensate previous steps
        workflow.ExecuteActivity(ctx, activities.RefundPayment, orderID).Get(ctx, nil)
        return nil
    }
    return workflow.ExecuteActivity(ctx, activities.FinalizePayment, orderID).Get(ctx, nil)
}

// To send approval signal from your API handler:
func ApprovePaymentHandler(c *gin.Context) {
    orderID := c.Param("id")
    temporalClient.SignalWorkflow(c, orderID, "", "approval-signal", true)
    c.JSON(200, gin.H{"status": "approved"})
}
Go / Temporal